package hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

public class UserLogFromHBase {
    // 放置初始化变量
    public Configuration config;

    public UserLogFromHBase() throws IOException {
        initConfig();
    }

    // 放置初始化方法
    public void initConfig() {
        config = HBaseConfiguration.create();
        config.set("hbase.zookeeper.quorum", "192.168.43.11");
    }

    /**
     * 获取指定RowKey对应的所有列的值，并通过控制台打印出来
     *
     * @param rowkey
     * @throws IOException
     */

    /**
     * 查询当前表中说有的用户兴趣日志（在真实业务系统中，不允许不带条件的全表扫描）
     *
     * @return 经过处理的用户兴趣日志
     * @throws IOException
     */
    public List<String> getUserLogList(String tableName, String[] columnFamilies) throws IOException {

        List<String> userlogList = new ArrayList();
        // 从HBase中获取表的所有数据，并返回格式化后的结果
        // 判断如果传入的参数不合规，直接返回空值
        if (columnFamilies == null || tableName.isEmpty() ||
                columnFamilies == null || columnFamilies.length == 0) {
            return userlogList;
        }
// 创建HBase的连接
        try (
                Connection connection = ConnectionFactory.createConnection(config);
                // 获取参数指定的表
                Table table = connection.getTable(TableName.valueOf(tableName));
        ) {
            // 创建Scan对象
            Scan scan = new Scan();
// 设置scan命令每次获取数据的条数
            scan.setBatch(100);


// 设置scan命令每次获取数据的条数

            // 添加参数指定的列族到Scan对象
            for( String cf:columnFamilies){
                scan.addFamily(Bytes.toBytes(cf));
            }


// 获取扫描的返回结果
            ResultScanner resultScanner = table.getScanner(scan);
            Result rs = resultScanner.next();


// 遍历查询到的所有行，并对行内的Cell进行遍历，获取列名和对应的值
            for (; rs != null;rs = resultScanner.next()){
                for ( Cell cell : rs.listCells()){
                    //RowKey
                    String rowkey = Bytes.toString(cell.getRowArray(),cell.getRowOffset(),cell.getRowLength())+":";
                    // 列族
                    String colFamily = Bytes.toString(cell.getFamilyArray(),cell.getFamilyOffset(),cell.getFamilyLength())+":";
                    //列
                    String column= Bytes.toString(cell.getQualifierArray(),cell.getQualifierOffset(),cell.getQualifierLength())+"=";
                    //值
                    String value = Bytes.toString(cell.getValueArray(),cell.getValueOffset(),cell.getValueLength());
                    userlogList.add(rowkey+colFamily+column+value);
                    System.out.println(rowkey+colFamily+column+value);
                }
            }

        }

        return userlogList;
    }

}
